package com.duowan.realtime.computing;

import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool.BasePoolableObjectFactory;
import org.apache.commons.pool.ObjectPool;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import com.duowan.realtime.thirft.api.Constants;
import com.duowan.realtime.thirft.api.HyperLogLogException;
import com.duowan.realtime.thirft.api.HyperLogLogPlusException;
import com.duowan.realtime.thirft.api.HyperLogLogQuery;
import com.duowan.realtime.thirft.api.HyperLogLogService;
import com.duowan.realtime.thirft.api.HyperLogLogService.Client;
import com.duowan.realtime.thirft.api.HyperLogLogService.Iface;

public class RawHyperLogLogClient implements Iface,InitializingBean,DisposableBean {
	private String host;
	private int port = Constants.DEFAULT_HLL_SERVER_PORT;
	
	private int clientPoolSize = 2;
	
	private final static Logger LOG = LoggerFactory.getLogger(RawHyperLogLogClient.class);
	
	private ObjectPool<HyperLogLogService.Client> clientPool;
	
	public RawHyperLogLogClient(){
	}
	
	public String getHost() {
		return host;
	}

	public void setHost(String host) {
		this.host = host;
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}
	
	@Override
	public String ping() throws HyperLogLogPlusException, TException {
		Client client = borrowObject();
		try {
			return client.ping();
		}finally {
			returnObject(client);
		}
	}

	/**
	 * 此方法用于计算历史新用户
	 * @param hllGroup hyperloglog组
	 * @param hllQueryList	批量的hyperloglogplus model
	 * @return values在该group下含有多少个新增
	 */
	@Override
	public Map<String, Integer> offerForCardinality(String hllGroup, List<HyperLogLogQuery> hllQueryList) throws HyperLogLogPlusException, TException {
		return doComputing(hllGroup, hllQueryList);
	}

	private Map<String, Integer> doComputing(String hllpGroup, List<HyperLogLogQuery> hllpQueryList) {
		Client client = borrowObject();
		try {
			Map<String, Integer> resultMap = client.offerForCardinality(hllpGroup,hllpQueryList);
			returnObject(client);
			return resultMap;
		} catch (HyperLogLogPlusException e) {
			returnObject(client);
			throw new RuntimeException("error on run notContainsCountAndAdd method:", e);
		} catch (TException e) {
			invalidateObject(client);
			throw new RuntimeException("error on run notContainsCountAndAdd method:", e);
		}
	}
	
	
	@Override
	public void destroy() throws Exception {
		if(clientPool != null) {
			clientPool.close();
		}
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		if(StringUtils.isBlank(host)) throw new IllegalStateException("host must be not empty");
		if(port <= 0) throw new IllegalArgumentException("port > 0 must be true");
		if(clientPoolSize <= 0) throw new IllegalArgumentException("clientPoolSize > 0 must be true");
		
		clientPool = new GenericObjectPool<HyperLogLogService.Client>(new ClientPoolableObjectFactory(),clientPoolSize);
		LOG.info("init end,server="+host+":"+port+" clientPoolSize:" + clientPoolSize);
	}
	
	private void invalidateObject(Client client) {
		try {
			clientPool.invalidateObject(client);
		} catch (Exception e) {
			throw new RuntimeException("invalidateObject error",e);
		}
	}
	
	private void returnObject(Client client) {
		try {
			clientPool.returnObject(client);
		} catch (Exception e) {
			throw new RuntimeException("returnObject error",e);
		}
	}
	
	private Client borrowObject() {
		try {
			return clientPool.borrowObject();
		} catch (Exception e) {
			throw new RuntimeException("borrowObject error",e);
		}
	}

	private class ClientPoolableObjectFactory extends BasePoolableObjectFactory<HyperLogLogService.Client> {
		final Map<Client,TTransport> clientTTransportMap = new Hashtable<HyperLogLogService.Client, TTransport>();
		@Override
		public HyperLogLogService.Client makeObject() throws Exception {
			TTransport transport = new TSocket(host, port);
			TProtocol protocol = new TBinaryProtocol(transport);
			transport.open();
			HyperLogLogService.Client client = new HyperLogLogService.Client(protocol);
			
			Assert.isTrue(validateObject(client),"client ping() error");
			
			clientTTransportMap.put(client, transport);
			LOG.info("connected_to_server:"+host+":"+port+" clientPool.numActive:"+clientPool.getNumActive()+" clientPool.numIdle:"+clientPool.getNumIdle()+" clientTTransportMap.size:"+clientTTransportMap.size());
			return client;
		}
		
		@Override
		public void destroyObject(HyperLogLogService.Client obj) throws Exception {
			TTransport transport = clientTTransportMap.remove(obj);
			if(transport != null) {
				LOG.info("destroyObject() closed_transport, server:"+host+":"+port+" clientPool.numActive:"+clientPool.getNumActive()+" clientPool.numIdle:"+clientPool.getNumIdle()+" clientTTransportMap.size:"+clientTTransportMap.size());
				transport.close();
			}
		}
		
		@Override
		public boolean validateObject(Client obj) {
			String ping = null;
			try {
				ping = obj.ping();
				if(Constants.PING_RESPONSE.equals(ping)) {
					return true;
				}
				return false;
			} catch (Exception e) {
				ping = e.toString();
				return false;
			}finally {
				LOG.info("validateObject,BloomFilterService.Client ping() "+host+" and get response:"+ping);
			}
		}
	}

	@Override
	public void offer(String hllGroup, List<HyperLogLogQuery> hllQueryList) throws HyperLogLogException, TException {
		Client client = borrowObject();
		try {
			client.offer(hllGroup, hllQueryList);
			returnObject(client);
		} catch (HyperLogLogPlusException e) {
			returnObject(client);
			throw new RuntimeException("error on run offer method:", e);
		} catch (TException e) {
			invalidateObject(client);
			throw new RuntimeException("error on run offer method:", e);
		}
	}

	@Override
	public long cardinality(String hllGroup, String group) throws HyperLogLogException, TException {
		long result = 0L;
		Client client = borrowObject();
		try {
			result = client.cardinality(hllGroup, group);
			returnObject(client);
		} catch (HyperLogLogPlusException e) {
			returnObject(client);
			throw new RuntimeException("error on run cardinality method:", e);
		} catch (TException e) {
			invalidateObject(client);
			throw new RuntimeException("error on run cardinality method:", e);
		}
		return result;
	}

	@Override
	public List<Map<String, String>> keys(String hllGroup, String key) throws HyperLogLogException, TException {
		Client client = borrowObject();
		List<Map<String, String>> result = null;
		try {
			result = client.keys(hllGroup, key);
			returnObject(client);
		} catch (HyperLogLogPlusException e) {
			returnObject(client);
			throw new RuntimeException("error on run offer method:", e);
		} catch (TException e) {
			invalidateObject(client);
			throw new RuntimeException("error on run offer method:", e);
		}
		return result;
	}




}
